package com.flow;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * 定义一个流中间操作,给每一个元素加上一个haha前缀   只用写订阅者的接口,发布者的接口已经继承了
 * 中间处理器要和上一个发布者进行绑定
 */
public class MyProcessor extends SubmissionPublisher<String> implements Flow.Processor<String,String> {

    private Flow.Subscription subscription;  //保存绑定关系

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.println("Processor订阅绑定完成");
        this.subscription = subscription;
        subscription.request(1); //找上游要一个数据
    }

    @Override  //数据到达  触发这个回调
    public void onNext(String item) {
        System.out.println("Processor拿到数据" + item);
        //数据加工
        item = "haha" + item;
        submit(item);  //将加工后的数据发出去
        subscription.request(1); //再要一个数据,继续加工发出去
    }

    @Override
    public void onError(Throwable throwable) {

    }

    @Override
    public void onComplete() {

    }
}
